package com.markhsiu.store.disk;

import java.io.File;
import java.util.zip.DataFormatException;

import com.markhsiu.core.constant.Bits;
import com.markhsiu.main.CompressUtils;
import com.markhsiu.store.StoreDB;

import io.openmessaging.demo.DefaultBytesMessage;

/**
 * topic Segment 文件
 * 
 * 
 * @author Mark Hsiu
 *
 */
class TopicSegment {
			
	private MapFileRead readMap;
	private MapFileWrite writeMap;
	private ThreadLocal<Cache> CacheLocal = new ThreadLocal<>();

	public TopicSegment() {}
	
	public void builder(String fileName){
		File file = new File(StoreDB.getRootDir() + fileName);
		writeMap = new MapFileWrite(file);
	}

	public void load(File file){
		readMap = new MapFileRead(file);
	}

	
	public void putBuffM(DefaultBytesMessage message) {
		
		int len = message.len();
	
		Cache cache = getCache(); 
		
		if (Cache.BUFF - cache.offset < len) {
			writeBlock(cache);
			cache.clear();
		}
	
		MessageUtil.buildBytes(message, cache.buff, cache.offset);
		cache.offset += len;
	}
	
 
	//block :  MAGIC + len + (message * n)
	private  void writeBlock(Cache cache){
		 
		int count = CompressUtils.compress(cache.buff, 0, cache.offset, cache.temp, MapFile.CACHE_HEAD, cache.offset);
		
		cache.temp[0] = StoreDB.MAGIC;// index = 1
		Bits.putShort(cache.temp, 1, (short) count);// index = 1+2
		
		writeMap.write(cache.buffer, cache.temp, 0, count + MapFile.CACHE_HEAD);			
	}
	
	public Cache getCacheM() {
		Cache cache = getCache();
		if(cache.mark == cache.offset || cache.mark == -1 ){
			
			int len = readMap.readBlockM(cache.buffer, cache.temp);
			if(len == 0){
				return null;
			}
			
			try {
				cache.mark = CompressUtils.uncompress(cache.temp, MapFile.CACHE_HEAD, len, cache.buff);
				cache.offset = 0;
			} catch (DataFormatException e) {
				e.printStackTrace();
			}	
		} 
		return  cache;
	}

	public Cache getCache() {
	
		
		Cache cache = CacheLocal.get();
		if(cache == null){
			
			cache = new Cache();
			if(StoreDB.isProducer){
				cache.buffer = writeMap.duplicate();
			} else {
				cache.buffer = readMap.duplicate();
			}
			
			CacheLocal.set(cache);
			
		}
		
		return  cache;
	}

	public void flush() {
		 
		Cache cache = getCache();
		if(cache.offset > 0){
			writeBlock(cache);
			 cache.clear() ;
		}
		
//		long s1 = System.nanoTime();
//		long s2 = System.nanoTime();
//		time.addAndGet(s1-s2);
		
		
	 
		//System.out.println(writeMap.position());
//		System.out.println(Thread.currentThread().getName() + " seg  "+(time.get()/1000000) +" ms");
	}
	
	 
}
